package io.grpc.benchmarks.qps;

import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.SimpleRequest;
import io.grpc.testing.SimpleResponse;
import org.HdrHistogram.Histogram;

/**
 * This seems necessary as we need to reference the requestObserver in the
 * responseObserver.
 * The alternative would be to use the channel layer directly.
 */
public class ThisIsAHackStreamObserver
        implements StreamObserver<SimpleResponse> {

    final SimpleRequest request;
    final Histogram histogram;
    final HistogramFuture future;
    final long endTime;
    long lastCall = System.nanoTime();

    StreamObserver<SimpleRequest> requestObserver;

    ThisIsAHackStreamObserver(SimpleRequest request, Histogram histogram,
            HistogramFuture future, long endTime) {
        this.request = request;
        this.histogram = histogram;
        this.future = future;
        this.endTime = endTime;
    }

    @Override
    public void onNext(SimpleResponse value) {
        long now = System.nanoTime();
        // Record the latencies in microseconds
        histogram.recordValue((now - lastCall) / 1000);
        lastCall = now;

        if (endTime > now) {
            requestObserver.onNext(request);
        } else {
            requestObserver.onCompleted();
        }
    }

    @Override
    public void onError(Throwable t) {
        Status status = Status.fromThrowable(t);
        System.err.println(
            "Encountered an error in streamingCall. Status is " + status);
        t.printStackTrace();

        future.cancel(true);
    }

    @Override
    public void onCompleted() {
        future.done();
    }
}